package com.gis.bigdata.spark02.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author LnnuUser
 * @create 2022-10-05-下午10:11
 */
object Scala_Test01_2 {

  /*
  日期，用户id，session ID，页面id，动作时间，搜索关键字，点击品类id，产品id，下单品类id、支付品类id，产品id，城市id
  2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_48_2019-07-17 00:00:10_null_16_98_null_null_null_null_19
  2019-07-17,95,26070e87-1ad7-49a3-8fb3-cc741facaddf,48,2019-07-17 00:00:10,null,16,98,null,null,null,null,19
   */
  def main(args: Array[String]): Unit = {
    // 需求1：Top10热门品类
    // TODO: TOP10
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
    val sc: SparkContext = new SparkContext(sparkConf)

    // TODO 1. 读取原始数据
    val actionRDD: RDD[String] = sc.textFile("/home/lnnu/IdeaProjects/spark/datas/user_visit_action.txt")
    // Q1. RDD重复使用
    // 2. 将数据转换结构
    // 点击：（品类id， （1，0，0））
    // 下单：（品类id， （0，1，0））
    // 支付：（品类id， （0，0，1））
    val flatRDD: RDD[(String, (Int, Int, Int))] = actionRDD.flatMap(
      action => {
        val datas: Array[String] = action.split("_")
        if (datas(6) != "-1") {
          // 点击
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          // 下单
          val ids: Array[String] = datas(8).split(",")
          ids.map(id => (id, (0, 1, 0)))
        } else if (datas(10) != "null") {
          // 支付
          val ids: Array[String] = datas(10).split(",")
          ids.map(id => (id, (0, 0, 1)))
        } else {
          Nil
        }
      }
    )

    // 3. 将相同的品类id的数据进行分组聚合
    // （品类id， （点击，下单， 支付））
    val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey(
      (t1, t2) => {
        ((t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3))
      }
    )

    // 4. 将统计结果根据数量降序排列，取前10

    // 5. 将结果采集到控制台打印
    val resultRDD: Array[(String, (Int, Int, Int))] = analysisRDD.sortBy(_._2, false).take(10)

    resultRDD.foreach(println)

    sc.stop()
  }
}
